package se.kth.ict.id2203.abortableconsensus;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.AbstractMap.SimpleImmutableEntry;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import se.kth.ict.id2203.abortableconsensus.AbortableConsensus;
import se.kth.ict.id2203.abortableconsensus.AcPropose;
import se.kth.ict.id2203.beb.BebBroadcast;
import se.kth.ict.id2203.beb.BestEffortBroadcast;
import se.kth.ict.id2203.pp2p.PerfectPointToPointLink;
import se.kth.ict.id2203.pp2p.Pp2pSend;
import se.sics.kompics.ComponentDefinition;
import se.sics.kompics.Handler;
import se.sics.kompics.Negative;
import se.sics.kompics.Positive;
import se.sics.kompics.address.Address;
import se.sics.kompics.launch.Topology;


public class RWAbortableConsensus extends ComponentDefinition {
	
	Negative<AbortableConsensus> acPort = provides(AbortableConsensus.class);
	
	Positive<BestEffortBroadcast> beb = requires(BestEffortBroadcast.class);
	Positive<PerfectPointToPointLink> pp2p = requires(PerfectPointToPointLink.class);	
	
	private Address self;
	private Topology topology;
	
	private int N;
	
	private Set<Integer> seenIds;
	private int majority;
		
	/*
	 * If I did this algo again I'd probably create a private instance class
	 * that contains all of these variables instead of this ugly solution...
	 * but hey, it works =)
	 */
	private Map<Integer,Integer> tempValue;
	private Map<Integer,Integer> wAcks;
	private Map<Integer,Integer> rts;
	private Map<Integer,Integer> wts;
	private Map<Integer,Integer> tStamp;
	private Map<Integer,List<Map.Entry<Integer,Integer>>> readSet;
	
	
	
	/*
	 * it would be enough with a set since it is true/false but I'll
	 * keep this as a map just to be able to duplicate the algorithm closer.
	 */
	
	private static final Logger logger =
			LoggerFactory.getLogger(RWAbortableConsensus.class);
	
	public RWAbortableConsensus() {
		logger.info("RWAbortableConsensus created.");
		subscribe(handleInit, control);
		subscribe(handleAcPropose, acPort);
		subscribe(handleReadMessage, beb);
		subscribe(handleNackMessage, pp2p);
		subscribe(handleReadAckMessage, pp2p);
		subscribe(handleWriteMessage, beb);
		subscribe(handleWriteAckMessage, pp2p);
	}
	
	Handler<RWAbortableConsensusInit> handleInit = new Handler<RWAbortableConsensusInit>() {
		public void handle(RWAbortableConsensusInit event) {
			
			logger.info("RWAbortableConsensusInit started.");
			topology = event.getTopology();
			self = topology.getSelfAddress();
			
//			logger.info("self = "+self.getId());
			
			N = topology.getNodeCount();
			
			seenIds = new HashSet<Integer>();
			majority = (N/2)+1;
			
			tempValue = new HashMap<Integer,Integer>();
			wAcks = new HashMap<Integer,Integer>();
			rts = new HashMap<Integer,Integer>();
			wts = new HashMap<Integer,Integer>();
			tStamp = new HashMap<Integer,Integer>();
			
			readSet = new HashMap<Integer,List<Map.Entry<Integer,Integer>>>();

		}
	};
	
	/*
	 * procedure initInstance(id)
	 */
	public void initInstance( int id )
	{
		logger.info("initInstance");
		// if (id /∈ seenIds) then
		if ( !seenIds.contains( id ) )
		{
//			logger.info("( !seenIds.contains( id ) )");
			// tempValue[id] := val[id] := ⊥; // will be denoted as the map does not contain the value.
			// wAcks[id] := rts[id] := wts[id] := 0;
			wAcks.put(id, 0);
			rts.put(id, 0);
			wts.put(id, 0);
			
//			logger.info("Awts.size="+wts.size());
//			logger.info("Awts.get(id)="+wts.get(id));
			
			// tstamp[id] := rank(self)
			tStamp.put(id, self.getId());
			// readSet[id] := ∅;
			readSet.put( id, new ArrayList<Map.Entry<Integer,Integer>>() );
			// seenIds := seenIds ∪ {id};
			seenIds.add(id);

		}
	}
	
	/*
	 * Read phase
	 */
		
	/*
	 * upon event ⟨ acP ropose | id, v ⟩ do
	 */
	Handler<AcPropose> handleAcPropose = new Handler<AcPropose>() {
		public void handle(AcPropose event) 
		{
			logger.info("handleAcPropose");
			int id = event.getId();
			// initInstance(id);
			initInstance(id);
			// tstamp[id] := tstamp[id] + N ;
			tStamp.put(id, tStamp.get(id)+N );
			// tempValue[id] := v;
			tempValue.put(id, event.getProposal());
			// trigger ⟨ bebBroadcast | [Read, id, tstamp[id]] ⟩;
//			trigger(new ReadMessage(self,id,tStamp.get(id)), beb);
			trigger(new BebBroadcast(new ReadMessage(self,id,tStamp.get(id))), beb);

		}

	};
	
	
	/*
	 * upon event ⟨ bebDeliver | pj , [Read, id, ts] ⟩ do
	 */
	Handler<ReadMessage> handleReadMessage = new Handler<ReadMessage>() {
		public void handle(ReadMessage event) 
		{

			logger.info("handleReadMessage from "+event.getSource().getId());
			int id = event.getId();
//			int ts = tStamp.get(id);
			int ts = event.getTimeStamp();
			
			// initInstance(id);
			initInstance(id);
			
//			logger.info("Bwts.size="+wts.size());
//			logger.info("Bwts.get(id)="+wts.get(id));
			
//			logger.info("DEBUG: id="+id+", wts[id]="+wts.get(id)+" rts[id]="+rts.get(id)+" ts="+ts);
			
			// if (rts[id] ≥ ts or wts[id] ≥ ts) then
			if ( ( rts.get(id) >= ts ) || ( wts.get(id) >= ts ) )
			{
				// trigger ⟨ pp2pSend | pj , [Nack, id] ⟩;
//				trigger(new NackMessage(event.getSource(), id),	pp2p);
				trigger(new Pp2pSend(event.getSource(), new NackMessage(self, id)), pp2p);
			}
			// else
			else
			{
				// rts[id] := ts;
				rts.put(id, ts);
				// trigger ⟨ pp2pSend | pj , [ReadAck, id, wts[id], val[id], ts] ⟩;
//				trigger(new ReadAckMessage(event.getSource(), id, wts.get(id), tempValue.get(id), ts),	pp2p);
//				logger.info("DEBUG: trigger Pp2pSend to "+event.getSource().getId()+
//						", id="+id+", wts[id]="+wts.get(id)+" tempValue[id]="+tempValue.get(id)+" ts="+ts);
				trigger(new Pp2pSend(event.getSource(), new ReadAckMessage(self, id, wts.get(id), tempValue.get(id), ts)), pp2p);
			}

		}

	};
	
	/*
	 * Write Phase
	 */
	
	
	/* 
	 * upon event ⟨ pp2pDeliver | pj , [Nack, id] ⟩ do
	 */
	Handler<NackMessage> handleNackMessage = new Handler<NackMessage>() {
		public void handle(NackMessage event) 
		{
			logger.info("handleNackMessage from "+event.getSource().getId());
			int id = event.getId();
			// readSet[id] := ∅;
			readSet.get(id).clear();
			// wAcks[id] := 0;
			wAcks.put(id, 0);
			// trigger ⟨ acReturn | id, ⊥ ⟩;
			trigger(new AcReturn(id, null), acPort);			
			
		}
		
	};
	
	
	/*
	 * upon event ⟨ pp2pDeliver | pj , [ReadAck, id, ts, v, sentts] ⟩ do
	 */
	Handler<ReadAckMessage> handleReadAckMessage = new Handler<ReadAckMessage>() {
		public void handle(ReadAckMessage event) 
		{
			logger.info("handleReadAckMessage from "+event.getSource().getId());
			int id = event.getId();
			int ts = event.getTimeStamp();
			int sentts = tStamp.get(id);
			Integer val = event.getVal();
			
			logger.info("id="+id+", ts="+ts +", sentts="+sentts);
			
			// if (sentts = tstamp[id]) then
			if ( ts == sentts )
			{
				List<Map.Entry<Integer,Integer>> rSet = readSet.get(id);
				// readSet[id] := readSet[id] ∪ {(ts, v)};
				
//				rSet.add(ts, event.getVal());
				rSet.add( new SimpleImmutableEntry<Integer,Integer>(ts, event.getVal()) );
				
				logger.info("rSet.size="+rSet.size());
				
				// if (|readSet[id]| = majority) then
				if( rSet.size() == majority )
				{
					logger.info("got majority!");
					
					// (ts, v) := highest(readSet[id]);
					for ( Map.Entry<Integer,Integer> entry : rSet )
					{
						if ( entry.getKey() > ts )
						{
							ts = entry.getKey();
							val = entry.getValue();
						}
					}
					
					// if (v ̸= ⊥) then
					if( !tempValue.containsKey(id) )
					{
						// tempValue[id] := v;
						tempValue.put(id, val);
					}
					// trigger ⟨ bebBroadcast | [Write,id,tstamp[id],tempValue[id]] ⟩;
//					trigger(new WriteMessage(self,id,ts, tempValue.get(id)), beb);
					trigger(new BebBroadcast(new WriteMessage(self,id,ts, tempValue.get(id))), beb);
					
				}

			}

		}
		
	};

	
	
	/*
	 * upon event ⟨ bebDeliver | pj , [Write, id, ts, v] ⟩ do
	 */
	Handler<WriteMessage> handleWriteMessage = new Handler<WriteMessage>() {
		public void handle(WriteMessage event) 
		{
			logger.info("handleWriteMessage from "+event.getSource().getId());
			int id = event.getId();
			int ts = event.getTimeStamp();
			
			// initInstance(id);
			initInstance(id);
			// if (rts[id] > ts or wts[id] > ts) then
			if ( ( rts.get(id) > ts ) || ( wts.get(id) > ts ) )
			{
				// trigger ⟨ pp2pSend | pj , [Nack, id] ⟩;
//				trigger(new NackMessage(event.getSource(), id),	pp2p);
				trigger(new Pp2pSend(event.getSource(), new NackMessage(event.getSource(), id)), pp2p);
			}
			// else
			else
			{
				// val[id] := v;
				tempValue.put(id, event.getValue());
				// wts[id] := ts;
				wts.put(id, event.getTimeStamp());
				// trigger ⟨ pp2pSend | pj , [WriteAck, id, ts] ⟩;
//				trigger(new WriteAckMessage(event.getSource(), id, wts.get(id), tempValue.get(id), ts),	pp2p);
				trigger(new Pp2pSend(event.getSource(), new WriteAckMessage(self, id, wts.get(id), tempValue.get(id), ts)), pp2p);
			}

		}
		
	};

		
		
	/*
	 * upon event ⟨ pp2pDeliver | pj , [WriteAck, id, sentts] ⟩ do
	 */
	Handler<WriteAckMessage> handleWriteAckMessage = new Handler<WriteAckMessage>() {
		public void handle(WriteAckMessage event) 
		{
			logger.info("handleWriteAckMessage from "+event.getSource().getId());
			int id = event.getId();
			int sentts = tStamp.get(id);
			
			// if (sentts = tstamp[id]) then
			if ( sentts == tStamp.get(id) )
			{
				// wAcks[id] := wAcks[id] + 1;
				wAcks.put(id, wAcks.get(id)+1);
				// if (wAcks[id] = majority) then
				if( wAcks.get(id) == majority )
				{
					// readSet[id] := ∅; wAcks[id] := 0;
					readSet.get(id).clear();
					wAcks.put(id, 0);
					// trigger ⟨ acReturn | id, tempValue[id]] ⟩;
					trigger(new AcReturn(id, event.getVal()), acPort);
				}

			}

		}
		
	};




	
	
}
